package com.jaws.core.configuration;

import java.util.Map;
import java.util.stream.Collectors;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import javax.sql.DataSource;

import com.jaws.core.common.xxljob.executor.JawsJobExecutor;
import com.xxl.job.core.executor.XxlJobExecutor;
import org.apache.commons.lang3.StringUtils;
import org.apache.ibatis.plugin.Interceptor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.TaskExecutor;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import com.ctrip.framework.apollo.spring.annotation.EnableApolloConfig;
import com.dangdang.ddframe.job.api.simple.SimpleJob;
import com.dangdang.ddframe.job.config.JobCoreConfiguration;
import com.dangdang.ddframe.job.config.simple.SimpleJobConfiguration;
import com.dangdang.ddframe.job.event.rdb.JobEventRdbConfiguration;
import com.dangdang.ddframe.job.lite.api.listener.ElasticJobListener;
import com.dangdang.ddframe.job.lite.config.LiteJobConfiguration;
import com.dangdang.ddframe.job.lite.spring.api.SpringJobScheduler;
import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperConfiguration;
import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperRegistryCenter;
import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.jaws.core.common.cache.JawsStringSerializer;
import com.jaws.core.common.elasticjob.JawsSimpleJob;
import com.jaws.core.common.initialize.SnowflakeInitializer;
import com.jaws.core.common.log.JawsThreadDecorator;
import com.jaws.core.common.log.LogScanner;
import com.jaws.core.common.shard.algorithm.BaseShardAlgorithm;
import com.jaws.core.common.shard.interceptor.JawsShardSqlInterceptor;
import com.jaws.core.common.validator.JawsValidator;

/**
 * 通用配置，主要区别web
 *
 * @author maojian
 * @date 2018-03-01 17:38:36
 * @since v1.0.0
 */
@Configuration
@EnableConfigurationProperties(CommonProperties.class)
public class CommonConfiguration {

	@Resource
	CommonProperties commonProperties;

	@Resource
	ApplicationContext context;

	@Bean
	public LogScanner logScanner() {
		LogScanner scanner = new LogScanner();
		scanner.setSensLogPath(commonProperties.getLogSensPath());
		scanner.setSensLogSwitch(commonProperties.getLogSensSwitch());
		return scanner;
	}

	@Bean
	public JawsValidator jawsValidator() {
		return new JawsValidator();
	}

	@Bean
	@ConditionalOnProperty(prefix = "jaws.common", name = "snowflakeMode", matchIfMissing = false)
	public SnowflakeInitializer snowflakeInitializer() {
		return new SnowflakeInitializer();
	}

	@Bean
	@ConditionalOnMissingBean(TaskExecutor.class)
	public TaskExecutor taskExecutor() {
		ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
		executor.setTaskDecorator(new JawsThreadDecorator());
		// 设置核心线程数
		executor.setCorePoolSize(commonProperties.getTaskExecutor().getCorePoolSize());
		// 设置最大线程数
		executor.setMaxPoolSize(commonProperties.getTaskExecutor().getMaxPoolSize());
		// 设置队列容量
		executor.setQueueCapacity(commonProperties.getTaskExecutor().getQueueCapacity());
		// 设置默认线程名称
		executor.setThreadNamePrefix(commonProperties.getTaskExecutor().getThreadNamePrefix());
		// 等待所有任务结束后再关闭线程池
		executor.setWaitForTasksToCompleteOnShutdown(true);
		executor.initialize();
		return executor;
	}

	@Configuration
	@ConditionalOnClass(JedisConnectionFactory.class)
	protected class JawsRedisConfiguration {

		@Bean
		public JawsStringSerializer jawsStringSerializer() {
			return new JawsStringSerializer(commonProperties.getRedis());
		}

		@Bean(name = "stringRedisTemplate")
		public StringRedisTemplate getStringRedisTemplate(JedisConnectionFactory jedisConnectionFactory) {
			StringRedisTemplate stringRedisTemplate = new StringRedisTemplate();
			stringRedisTemplate.setKeySerializer(jawsStringSerializer());
			stringRedisTemplate.setConnectionFactory(jedisConnectionFactory);
			return stringRedisTemplate;
		}

		@Bean(name = "redisTemplate")
		public RedisTemplate<Object, Object> getRedisTemplate(JedisConnectionFactory jedisConnectionFactory) {
			RedisTemplate<Object, Object> redisTemplate = new RedisTemplate<Object, Object>();
			redisTemplate.setConnectionFactory(jedisConnectionFactory);
			// 使用Jackson2JsonRedisSerialize 替换默认序列化
			Jackson2JsonRedisSerializer<Object> jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer<Object>(Object.class);

			ObjectMapper objectMapper = new ObjectMapper();
			objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
			objectMapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);

			jackson2JsonRedisSerializer.setObjectMapper(objectMapper);

			// 设置value的序列化规则和 key的序列化规则
			redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer());
			redisTemplate.setKeySerializer(jawsStringSerializer());
			redisTemplate.afterPropertiesSet();

			return redisTemplate;
		}

	}

	@Configuration
	@ConditionalOnClass(EnableApolloConfig.class)
	@EnableApolloConfig
	protected static class JawsApolloConfiguration {

	}

	@Configuration
	@ConditionalOnClass(SimpleJob.class)
	protected static class JawsElasticJobConfiguration {

		@Value("${elastic-job.zookeeper.server-lists}")
		private String serverList;

		@Value("${elastic-job.zookeeper.namespace}")
		private String namespace;

		@Autowired
		private ApplicationContext applicationContext;

		@PostConstruct
		public void initElasticJob() {
			ZookeeperConfiguration zooConfig = new ZookeeperConfiguration(serverList, namespace);
			zooConfig.setSessionTimeoutMilliseconds(1000);
			zooConfig.setConnectionTimeoutMilliseconds(10000);
			ZookeeperRegistryCenter regCenter = new ZookeeperRegistryCenter(zooConfig);
			regCenter.init();
			Map<String, SimpleJob> map = applicationContext.getBeansOfType(SimpleJob.class);

			for (Map.Entry<String, SimpleJob> entry : map.entrySet()) {
				SimpleJob simpleJob = entry.getValue();
				JawsSimpleJob simpleJobAnnotation = simpleJob.getClass().getAnnotation(JawsSimpleJob.class);
				boolean override = simpleJobAnnotation.overwrite();
				boolean monitorExecution = simpleJobAnnotation.monitorExecution();
				int monitorPort = simpleJobAnnotation.monitorPort();
				String cron = StringUtils.defaultIfBlank(simpleJobAnnotation.cron(), simpleJobAnnotation.value());
				SimpleJobConfiguration simpleJobConfiguration = new SimpleJobConfiguration(JobCoreConfiguration
						.newBuilder(simpleJobAnnotation.jobName(), cron, simpleJobAnnotation.shardingTotalCount()).jobParameter(simpleJobAnnotation.jobParameter())
						.shardingItemParameters(simpleJobAnnotation.shardingItemParameters()).failover(true).misfire(true).description(simpleJobAnnotation.description()).build(), simpleJob.getClass()
						.getCanonicalName());
				LiteJobConfiguration liteJobConfiguration = LiteJobConfiguration.newBuilder(simpleJobConfiguration).overwrite(override).monitorExecution(monitorExecution).monitorPort(monitorPort)
						.build();

				String dataSourceRef = simpleJobAnnotation.dataSource();
				if (StringUtils.isNotBlank(dataSourceRef)) {

					if (!applicationContext.containsBean(dataSourceRef)) {
						throw new RuntimeException("not exist datasource [" + dataSourceRef + "] !");
					}

					DataSource dataSource = (DataSource) applicationContext.getBean(dataSourceRef);
					JobEventRdbConfiguration jobEventRdbConfiguration = new JobEventRdbConfiguration(dataSource);
					if (StringUtils.isNotEmpty(simpleJobAnnotation.listener())) {
						ElasticJobListener listener = applicationContext.getBean(simpleJobAnnotation.listener(), ElasticJobListener.class);
						SpringJobScheduler jobScheduler = new SpringJobScheduler(simpleJob, regCenter, liteJobConfiguration, jobEventRdbConfiguration, listener);
						jobScheduler.init();
					} else {
						SpringJobScheduler jobScheduler = new SpringJobScheduler(simpleJob, regCenter, liteJobConfiguration, jobEventRdbConfiguration);
						jobScheduler.init();
					}
				} else {
					if (StringUtils.isNotEmpty(simpleJobAnnotation.listener())) {
						ElasticJobListener listener = applicationContext.getBean(simpleJobAnnotation.listener(), ElasticJobListener.class);
						SpringJobScheduler jobScheduler = new SpringJobScheduler(simpleJob, regCenter, liteJobConfiguration, listener);
						jobScheduler.init();
					} else {
						SpringJobScheduler jobScheduler = new SpringJobScheduler(simpleJob, regCenter, liteJobConfiguration);
						jobScheduler.init();
					}
				}
			}
		}
	}

	@Configuration
	@EnableConfigurationProperties(ShardTableProperties.class)
	@ConditionalOnProperty(prefix = "shard", name = "enable", havingValue = "true")
	protected static class ShardTableConfiguration {
		@Bean
		public BaseShardAlgorithm baseShardAlgorithm() {
			return new BaseShardAlgorithm();
		}

		@Bean
		public Interceptor shardTableInterceptor(ShardTableProperties shardTableProperties) {
			Map<String, ShardTableProperties.TableRule> tableRuleMap = shardTableProperties.getRules().stream().collect(Collectors.toMap(ShardTableProperties.TableRule::getTable, (p) -> p));
			return new JawsShardSqlInterceptor(tableRuleMap);
		}
	}

	@Configuration
	@ConditionalOnClass(XxlJobExecutor.class)
	protected static class JawsJobConfiguration {
		@Bean(initMethod = "start", destroyMethod = "destroy")
		public JawsJobExecutor jawsJobExecutor() {
			JawsJobExecutor jawsJobExecutor = new JawsJobExecutor();
			return jawsJobExecutor;
		}
	}

}
